package stc.skymobi.netty.transport.endpoint;

import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import stc.skymobi.netty.filter.Filter;
import stc.skymobi.netty.transport.Validateable;
import stc.skymobi.transport.protocol.esb.signal.HeartbeatToEsbReq;
import stc.skymobi.transport.protocol.esb.signal.HeartbeatToEsbResp;
import stc.skymobi.transport.protocol.esb.signal.RegisterToEsbReq;
import stc.skymobi.transport.protocol.esb.signal.RegisterToEsbResp;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public class ESBModuleEndpoint extends AbstractEndpoint
        implements MutableEndpoint {

    private static final Logger logger =
            LoggerFactory.getLogger(ESBModuleEndpoint.class);

    private List<Filter> filterList = Collections.emptyList();

    private ScheduledExecutorService exec =
            Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {

                public Thread newThread(Runnable r) {
                    return new Thread(r, "ESBModule-Sender-thread");
                }
            });

    private int moduleId;

    private int instanceId;

    //	esbConnectionState enum
    private static final int CONN_STATE_OFFLINE = 0;
    private static final int CONN_STATE_REGISTERING = 1;
    private static final int CONN_STATE_ONLINE = 2;

    //	esbHeartbeatState enum
    private static final int HB_STATE_IDLE = 0;
    private static final int HB_STATE_SENDED = 1;
    private static final int HB_STATE_RECVD = 2;

    private int esbConnectionState = CONN_STATE_OFFLINE;
    private int esbHeartbeatState = HB_STATE_IDLE;

    private long checkESBDelay = 5 * 1000;
    private long esbRespTimeout = 10 * 1000;
    private long esbHeartbeatInterval = 120 * 1000;
    private long fetchSignalTimeout = 1000;
    private long lastRegisterToEsbTime;
    private int reRegisterCount = 0;
    private long lastHeartbeatToEsbTime = 0;
    private long esbHeartbeatTimeout = 10 * 1000;
    private int reHeartbeatCount = 0;

    private void doSendPending() {
        exec.submit(new Runnable() {

            public void run() {
                sendPending();
            }
        });
    }

    private boolean isESBConnectionValid() {
        return ((null != channel) && (CONN_STATE_ONLINE == esbConnectionState));
    }

    private void sendPending() {
        if (isESBConnectionValid()) {
            try {
                Object bean = pendings.poll(fetchSignalTimeout, TimeUnit.MILLISECONDS);
                if (null != bean) {
                    channel.writeAndFlush(bean);
                }
            } catch (InterruptedException e) {
                logger.error("sendPending", e);
            } finally {
                doSendPending();
            }
        } else {
            //	esb not connected or not registered
            exec.schedule(new Runnable() {

                public void run() {
                    sendPending();
                }
            }, fetchSignalTimeout, TimeUnit.MILLISECONDS);
        }
    }

    private boolean isHeartbeatToEsbTimeout() {
        long duration = System.currentTimeMillis() - lastHeartbeatToEsbTime;
        return (duration >= esbHeartbeatTimeout);
    }

    private boolean isHeartbeatToEsbExceedInterval() {
        long duration = System.currentTimeMillis() - lastHeartbeatToEsbTime;
        return (duration >= esbHeartbeatInterval);
    }

    private void sendHeartbeatToEsb() {
        lastHeartbeatToEsbTime = System.currentTimeMillis();
        HeartbeatToEsbReq req = new HeartbeatToEsbReq();

        req.setDstESBAddr(0xE000);
        req.setSeqNum(0);

        channel.writeAndFlush(req);
    }

    private void doProcessHeartbeat() {
        if (HB_STATE_IDLE == this.esbHeartbeatState) {
            if (isHeartbeatToEsbExceedInterval()) {
                esbHeartbeatState = HB_STATE_SENDED;
                sendHeartbeatToEsb();
            }
        } else if (HB_STATE_SENDED == this.esbHeartbeatState) {
            if (isHeartbeatToEsbTimeout()) {
                if (reHeartbeatCount < 1) {
                    reHeartbeatCount++;
                    sendHeartbeatToEsb();
                } else {
                    //	heartbeat timeout for twice, disconnect session
                    this.channel.close();
                }
            }
        } else if (HB_STATE_RECVD == this.esbHeartbeatState) {
            reHeartbeatCount = 0;
            esbHeartbeatState = HB_STATE_IDLE;
        } else {
            throw new RuntimeException("Internal Error: Unknown ESB Heartbeat State ["
                    + esbHeartbeatState + "]");
        }
    }

    private boolean isRegisterToEsbTimeout() {
        long duration = System.currentTimeMillis() - lastRegisterToEsbTime;
        return (duration >= esbRespTimeout);
    }

    private void sendRegisterToEsb() {
        lastRegisterToEsbTime = System.currentTimeMillis();
        RegisterToEsbReq req = new RegisterToEsbReq();

        req.setDstESBAddr(0xE000);
        req.setSeqNum(0);

        //	由于现网有些模块不支持发送心跳，为了区分，如果模块支持心跳，
        //	在发送第一个注册消息时，请将ESB消息头中的FLAG设置为1。
        req.setFlags(1);

        req.setModuleId(this.moduleId);
        req.setInstanceId(this.instanceId);

        channel.writeAndFlush(req);
    }

    private void doCheckESB() {
        if (CONN_STATE_OFFLINE == esbConnectionState) {
            esbConnectionState = CONN_STATE_REGISTERING;
            sendRegisterToEsb();
        } else if (CONN_STATE_REGISTERING == esbConnectionState) {
            //	check registering timeout
            if (isRegisterToEsbTimeout()) {
                reRegisterCount++;
                //	register to esb timeout, try re-register
                sendRegisterToEsb();
            }
        } else if (CONN_STATE_ONLINE == esbConnectionState) {
            //	check weather to send heartbeat
            doProcessHeartbeat();
        } else {
            throw new RuntimeException("Internal Error: Unknown ESB Connection State ["
                    + esbConnectionState + "]");
        }
    }

    public void start() {
        exec.scheduleWithFixedDelay(new Runnable() {

            public void run() {
                doCheckESB();
            }
        },
                0, checkESBDelay, TimeUnit.MILLISECONDS);

        doSendPending();
    }

    public void stop() {
        super.stop();
        this.exec.shutdownNow();
    }

    /**
     * @param instanceId the instanceId to set
     */
    public void setInstanceId(int instanceId) {
        this.instanceId = instanceId;
    }

    /* (non-Javadoc)
      * @see com.skymobi.transport.AbstractEndpoint#messageReceived(org.apache.mina.core.session.IoSession, java.lang.Object)
      */
    @Override
    public void messageReceived(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof RegisterToEsbResp) {
            RegisterToEsbResp resp = (RegisterToEsbResp) msg;
            if (0 == resp.getResult()) {
                esbConnectionState = CONN_STATE_ONLINE;
                //	succees
                if (logger.isDebugEnabled()) {
                    logger.debug("register to esb with moduleId:"
                            + this.moduleId + ", instanceId:" + this.instanceId
                            + " succeed.");
                }
            } else {
                logger.warn("register to esb with moduleId:"
                        + this.moduleId + ", instanceId:" + this.instanceId
                        + " FAILED, result:" + resp.getResult());
            }
        } else if (msg instanceof HeartbeatToEsbResp) {
            esbHeartbeatState = HB_STATE_RECVD;
            if (logger.isDebugEnabled()) {
                logger.debug("recvd heartbeat to esb: {}", msg);
            }
        } else {
            for (Filter filter : filterList) {
                if (!filter.doFilter(ctx, msg)) { //只要有任何验证失败的操作,就不需要继续传递消息
                    return;
                }
            }
            if (msg instanceof Validateable) {
                Object result = ((Validateable) msg).validate();
                if (result != null) {
                    ctx.channel().writeAndFlush(result);
                    logger.info("request {} validate failed,return resp {}", msg, result);
                    return;
                }
            }
            super.messageReceived(ctx, msg);
        }
    }

    /**
     * @return the moduleId
     */
    public int getModuleId() {
        return moduleId;
    }

    /**
     * @param moduleId the moduleId to set
     */
    public void setModuleId(int moduleId) {
        this.moduleId = moduleId;
    }

    /**
     * @return the instanceId
     */
    public int getInstanceId() {
        return instanceId;
    }

    /**
     * @return the fetchSignalTimeout
     */
    public long getFetchSignalTimeout() {
        return fetchSignalTimeout;
    }

    /**
     * @param fetchSignalTimeout the fetchSignalTimeout to set
     */
    public void setFetchSignalTimeout(long fetchSignalTimeout) {
        this.fetchSignalTimeout = fetchSignalTimeout;
    }

    /**
     * @return the checkESBDelay
     */
    public long getCheckESBDelay() {
        return checkESBDelay;
    }

    /**
     * @param checkESBDelay the checkESBDelay to set
     */
    public void setCheckESBDelay(long checkESBDelay) {
        this.checkESBDelay = checkESBDelay;
    }

    /**
     * @return the esbHeartbeatInterval
     */
    public long getEsbHeartbeatInterval() {
        return esbHeartbeatInterval;
    }

    /**
     * @param esbHeartbeatInterval the esbHeartbeatInterval to set
     */
    public void setEsbHeartbeatInterval(long esbHeartbeatInterval) {
        this.esbHeartbeatInterval = esbHeartbeatInterval;
    }

    /**
     * @return the esbRespTimeout
     */
    public long getEsbRespTimeout() {
        return esbRespTimeout;
    }

    /**
     * @param esbRespTimeout the esbRespTimeout to set
     */
    public void setEsbRespTimeout(long esbRespTimeout) {
        this.esbRespTimeout = esbRespTimeout;
    }

    /**
     * @return the reRegisterCount
     */
    public int getReRegisterCount() {
        return reRegisterCount;
    }

    /**
     * @return the esbConnectionState
     */
    public int getEsbConnectionState() {
        return esbConnectionState;
    }

    /**
     * @return the esbHeartbeatState
     */
    public int getEsbHeartbeatState() {
        return esbHeartbeatState;
    }

    /**
     * @return the esbHeartbeatTimeout
     */
    public long getEsbHeartbeatTimeout() {
        return esbHeartbeatTimeout;
    }

    /**
     * @param esbHeartbeatTimeout the esbHeartbeatTimeout to set
     */
    public void setEsbHeartbeatTimeout(long esbHeartbeatTimeout) {
        this.esbHeartbeatTimeout = esbHeartbeatTimeout;
    }

    public void setFilterList(List<Filter> filterList) {
        this.filterList = filterList;
    }
}
